package com.exd.flink.sink.mysql;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions.Builder;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;

import com.exd.flink.common.PropertiesTool;
import com.exd.flink.domain.AlgorithmStatusBean;
import com.exd.flink.domain.VersionBean;
import com.exd.flink.sink.business.AlgorithmStatusDataSink;
import com.exd.flink.sink.business.VsersionDataSink;

public class MysqlSink {

	/**
	 * 版本数据插入
	 */
	public static void inserVersionData(DataStream<VersionBean> versionStream) {
		String sql = "";
		versionStream.addSink(JdbcSink.sink(sql,new VsersionDataSink(),getJdbcExecution(),getJdbcConnection()));
	}
	
	/**
	 * 修改算法状态
	 */
	public static void updateAlgorithmStatusData(DataStream<AlgorithmStatusBean> algorithmStatusStream) {
		String sql = "";
		algorithmStatusStream.addSink(JdbcSink.sink(sql,new AlgorithmStatusDataSink(),getJdbcExecution(),getJdbcConnection()));
	}
	
	/**
	 * MYSQL批量配置
	 */
	public static JdbcExecutionOptions getJdbcExecution() {
		Builder builder = JdbcExecutionOptions.builder();
		//批量写入的数据大小
		builder.withBatchSize(10);
		//每次批量写入最小间隔时间
		builder.withBatchIntervalMs(1000);
		// 插入发生异常重试次数
		builder.withMaxRetries(3);
		return builder.build();
	}
	
	/**
	 * MYSQL链接配置
	 */
	public static JdbcConnectionOptions getJdbcConnection() {
		JdbcConnectionOptions.JdbcConnectionOptionsBuilder jdbcConnectionOptionsBuilder = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder();
		ParameterTool parameterTool = PropertiesTool.getParameterTool();
		jdbcConnectionOptionsBuilder.withUrl(parameterTool.get("exd.mysql.url"));
		jdbcConnectionOptionsBuilder.withDriverName(parameterTool.get("exd.mysql.driver.name"));
		jdbcConnectionOptionsBuilder.withUsername(parameterTool.get("exd.mysql.username"));
		jdbcConnectionOptionsBuilder.withPassword(parameterTool.get("exd.mysql.password"));
		return jdbcConnectionOptionsBuilder.build();
	}
}
